package org.haifi.game.channel;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.haifi.game.reflection.GameReflect;
import org.haifi.game.util.SessionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

public class GameExecutorRouter {
	/**
	 * 全服单线程消息路由器
	 * 
	 * key:指令id
	 * 
	 * value:线程执行器
	 */
	private Map<Short, ThreadPoolTaskExecutor> router = Maps.newHashMap();

	/**
	 * 读并发指令和全服单线程指令
	 */
	private Set<Short> simpleCmds = Sets.newHashSet();

	/**
	 * 战斗指令:
	 * 
	 * TODO:可以改为单map单线程，提高并发量
	 */
	private Set<Short> battleCmds = Sets.newHashSet();

	/**
	 * 用户队列指令，单用户单线程，多用户并发
	 */
	private Set<Short> userOrderedCmds = Sets.newHashSet();

	/**
	 * 单用户单线程消息队列
	 * 
	 * key:gameUser.id
	 * 
	 */
	private Map<Integer, OrderedTask> userOrderedTasks = Maps
			.newConcurrentMap();

	/**
	 * 战斗消息队列,单地图单线程
	 * 
	 * key:map.id
	 */
	private Map<Integer, OrderedTask> battleOrderedTasks = Maps
			.newConcurrentMap();

	public void execute(Short cmdId, GameProcesserRunnable runnable) {

		// 只读并发指令或全服单线程指令
		if (simpleCmds.contains(cmdId)) {
			router.get(cmdId).execute(runnable);
			return;
		}

		ThreadPoolTaskExecutor executor = null;
		Integer key = null;
		Map<Integer, OrderedTask> tasks = null;
		// 用户队列指令
		if (userOrderedCmds.contains(cmdId)) {
			executor = userOrderedTaskExecutor;
			key = (Integer) runnable.ioSession
					.getAttribute(SessionUtils.USER_ID);
			tasks = userOrderedTasks;
		} else if (battleOrderedTasks.containsKey(cmdId)) {
			executor = battleTaskExecutor;
			// TODO:key=map.id
			tasks = battleOrderedTasks;
		} else {
			log.error("error cmdId:" + cmdId);
		}

		// 同步块放在if条件里，可以避免绝大部分语句进行同步
		OrderedTask task = tasks.get(key);
		if (task == null) {
			synchronized (tasks) {
				// 进入同步块后，再重新去获取一下，避免第二个线程把设置的属性给冲掉
				task = tasks.get(key);
				if (task == null) {
					task = new OrderedTask(executor);
					tasks.put(key, task);
				}
			}
		}
		task.add(runnable);
	}

	/**
	 * 初始化router列表
	 */
	public void init() {
		// 这里还需要处理一种全单线程的指令,添加完成
		// 所有指令
		List<Short> allCmds = Lists.newArrayList(GameReflect.getAllCmdIds());

		// 用户队列指令
		init(userOrderedCmd, userOrderedCmds, allCmds);
		// 战斗指令
		init(battleCmd, battleCmds, allCmds);

		// 全服单线程指令
		for (String cmds : initMap.keySet()) {
			ThreadPoolTaskExecutor executor = initMap.get(cmds);
			for (String cmd : cmds.split(",")) {
				Short cmdId = Short.valueOf(cmd);
				allCmds.remove(cmdId);
				simpleCmds.add(cmdId);
				router.put(cmdId, executor);
			}
		}

		// 把剩余所的未指定指令，全归入只读线程池
		for (Short cmdId : allCmds) {
			simpleCmds.add(cmdId);
			router.put(cmdId, readTaskExecutor);
		}
	}

	private void init(List<String> cmds, Set<Short> cmdSet, List<Short> allCmds) {
		for (String cmd : cmds) {
			for (String c : cmd.split(",")) {
				Short cmdId = Short.valueOf(c);
				cmdSet.add(cmdId);
				allCmds.remove(cmdId);
			}
		}
	}

	private class OrderedTask implements Runnable {

		private ThreadPoolTaskExecutor executor;
		private final Lock lock;
		private final Queue<Runnable> queue;

		OrderedTask(ThreadPoolTaskExecutor executor) {
			this.lock = new ReentrantLock();
			this.queue = new LinkedList<Runnable>();
			this.executor = executor;
		}

		public void add(Runnable runnable) {
			boolean runTask;
			lock.lock();
			try {
				// 消息队列为空时，允许立即丢入线程池
				runTask = queue.isEmpty();
				queue.offer(runnable);
			} finally {
				lock.unlock();
			}
			if (runTask) {
				executor.execute(this);
			}
		}

		@Override
		public void run() {
			Runnable runnable;
			boolean runAgain;
			lock.lock();
			try {
				runnable = queue.poll();
			} finally {
				// 如果消息队列还有更多的元素，则重新进入线程池
				runAgain = !queue.isEmpty();
				lock.unlock();
			}
			try {
				runnable.run();
			} catch (Exception ex) {
				ex.printStackTrace();
			}
			if (runAgain) {
				// 重新进入线程池执行
				executor.execute(this);
			}
		}
	}

	/**
	 * 清理掉空的orderedTasks
	 */
	public void cleanEmptyTasks() {
		// 清理掉的数量
		int count = 0;

		for (Integer uid : userOrderedTasks.keySet()) {
			if (!SessionUtils.isOnline(uid)) {
				OrderedTask task = userOrderedTasks.get(uid);
				if (task.queue.isEmpty()) {
					userOrderedTasks.remove(uid);
					count++;
					if (log.isDebugEnabled()) {
						log.debug("remove offline user tasks with uid:" + uid);
					}
				} else {
					log.error("user ordered tasks is not empty with uid:" + uid
							+ "\tsize:" + task.queue.size());
				}
			}
		}

		// 
		if (log.isInfoEnabled()) {
			log.info("clean empty tasks size:" + count);
		}

		// TODO:是否有空的map需要清理
	}

	public void setUserOrderedCmd(List<String> userOrderedCmd) {
		this.userOrderedCmd = userOrderedCmd;
	}

	public void setBattleCmd(List<String> battleCmd) {
		this.battleCmd = battleCmd;
	}

	public void setInitMap(Map<String, ThreadPoolTaskExecutor> initMap) {
		this.initMap = initMap;
	}

	/**
	 * 全服单线程指令
	 */
	private Map<String, ThreadPoolTaskExecutor> initMap;
	/**
	 * 配置的用户队列指令
	 */
	private List<String> userOrderedCmd;

	/**
	 * 配置的战斗指令
	 */
	private List<String> battleCmd;
	/**
	 * 默认的，只读线程池,多线程完全并发
	 */
	@Autowired
	private ThreadPoolTaskExecutor readTaskExecutor;

	/**
	 * 单用户单线程
	 */
	@Autowired
	private ThreadPoolTaskExecutor userOrderedTaskExecutor;

	/**
	 * 战斗线程池,全部唯一线程
	 * 
	 * TODO:未来要改为支持单地图单线程，可以提高并发量
	 */
	@Autowired
	private ThreadPoolTaskExecutor battleTaskExecutor;

	private static final Log log = LogFactory.getLog(GameExecutorRouter.class);
}
